package com.example.coupon.service.impl;

import com.alibaba.fastjson.JSON;
import com.example.coupon.constant.Constant;
import com.example.coupon.constant.CouponStatus;
import com.example.coupon.dao.CouponDao;
import com.example.coupon.entity.Coupon;
import com.example.coupon.exception.CouponException;
import com.example.coupon.feign.SettlementClient;
import com.example.coupon.feign.TemplateClient;
import com.example.coupon.service.IRedisService;
import com.example.coupon.service.IUserService;
import com.example.coupon.vo.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 用户服务相关的接口实现
 * 所有的操作过程，状态都保存在redis中，并通过kafka把消息传递到Mysql中
 *
 * @author wangbo
 * @description:
 * @date 2021/2/1914:51
 */
@Slf4j
@Service
public class UserServiceImpl implements IUserService {

    @Resource
    private CouponDao couponDao;

    @Resource
    private IRedisService redisService;

    @Resource
    private TemplateClient templateClient;

    @Resource
    private SettlementClient settlementClient;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 根据用户id和状态查询优惠券记录
     *
     * @param userId 用户id
     * @param status 优惠券状态
     * @return {@link Coupon}s
     * @throws CouponException
     */
    @Override
    public List<Coupon> findCouponsByStatus(Long userId, Integer status) throws CouponException {
        List<Coupon> curCached = redisService.getCacheCoupons(userId, status);
        List<Coupon> preTarget;
        if (CollectionUtils.isNotEmpty(curCached)) {
            log.debug("coupon cache is not empty: {},{}", userId, status);
            preTarget = curCached;
        } else {
            log.debug("coupon cache is empty,get coupon from db:{},{}", userId, status);
            List<Coupon> dbCoupons = couponDao.findAllByUserIdAndStatus(userId, CouponStatus.of(status));
            //如果数据库中没有记录，直接返回就可以
            if (CollectionUtils.isEmpty(dbCoupons)) {
                log.debug("current user do not have coupon:{},{}", userId, status);
                return dbCoupons;
            }
            //填充dbCoupons的templateSDK字段
            Map<Integer, CouponTemplateSDK> id2TemplateSDK = templateClient.findIds2TemplateSDK(dbCoupons.stream().map(Coupon::getTemplateId).collect(Collectors.toList())).getData();
            dbCoupons.forEach(dc -> dc.setCouponTemplateSDK(id2TemplateSDK.get(dc.getTemplateId())));
            //数据库中存在记录
            preTarget = dbCoupons;
            //将记录写入Cache
            redisService.addCouponToCache(userId, preTarget, status);
        }
        //将无效的优惠券剔除
        preTarget = preTarget.stream().filter(c->c.getId() != -1).collect(Collectors.toList());
        //如果当前获取的是可用优惠券，还需要做对已过期优惠券的延迟处理
        if(CouponStatus.of(status) == CouponStatus.USABLE){
            CouponClassify classify = CouponClassify.classify(preTarget);
            //如果已过期状态不为空，需要做延迟处理
            if (CollectionUtils.isNotEmpty(classify.getExpired())){
                log.info("Add Expired Coupons To Cache From FindCouponByStatus:{},{}",userId,status);
                redisService.addCouponToCache(userId,classify.getExpired(),CouponStatus.EXPIRED.getCode());
            }
            //发送到kafka中做异步处理
            kafkaTemplate.send(Constant.TOPIC, JSON.toJSONString(new CouponKafkaMessage(CouponStatus.EXPIRED.getCode(),classify.getExpired().stream().map(Coupon::getId).collect(Collectors.toList()))));
            return classify.getUsable();
        }
        return preTarget;
    }

    /**
     * 根据用户id 查找当前可以领取的优惠券模版
     *
     * @param userId 用户id
     * @return {@link CouponTemplateSDK}s
     * @throws CouponException
     */
    @Override
    public List<CouponTemplateSDK> findAvailableTemplate(Long userId) throws CouponException {
        long curTime = new Date().getTime();
        List<CouponTemplateSDK> templateSDKS = templateClient.findAllUsableTemplate().getData();

        log.debug("Find All Template(From TemplateClient) Count:{}",templateSDKS.size());
        //过滤过期的优惠券模版
        templateSDKS = templateSDKS.stream().filter(t->t.getTemplateRule().getExpiration().getDeadline() > curTime).collect(Collectors.toList());
        log.info("Find Usable Template Count:{}",templateSDKS.size());

        Map<Integer, Pair<Integer, CouponTemplateSDK>> limit2Template = new HashMap<>(templateSDKS.size());
        templateSDKS.forEach(t->limit2Template.put(t.getId(),Pair.of(t.getTemplateRule().getLimitation(),t)));

        List<CouponTemplateSDK> result = new ArrayList<>(limit2Template.size());
        List<Coupon> userUsableCoupons = findCouponsByStatus(userId,CouponStatus.USABLE.getCode());

        log.debug("Current User Has Usable Coupons:{},{}",userId,userUsableCoupons.size());

        Map<Integer, List<Coupon>> templateId2Coupons = userUsableCoupons.stream().collect(Collectors.groupingBy(Coupon::getTemplateId));

        //根据Template的Rule判断是否可以领取优惠券模版
        limit2Template.forEach((k, v)->{
            int limitation = v.getLeft();
            CouponTemplateSDK templateSDK = v.getRight();
            if (templateId2Coupons.containsKey(k)&&templateId2Coupons.get(k).size()>= limitation){
                return;
            }
            result.add(templateSDK);
        });

        return result;
    }

    /**
     * 用户领取优惠券
     * 1.从TemplateClient拿到对应的优惠券，并检查是否过期
     * 2.根据limitation判断用户是否可以领取
     * 3.save to db
     * 4.填充CouponTemplateSDK
     * 5.save to cache
     * @param request {@link AcquireTemplateRequest}
     * @return {@link Coupon}
     * @throws CouponException
     */
    @Override
    public Coupon acquireTemplate(AcquireTemplateRequest request) throws CouponException {
       Map<Integer, CouponTemplateSDK> id2Template = templateClient.findIds2TemplateSDK(Collections.singletonList(request.getTemplateSDK().getId())).getData();

       //优惠券模版是需要存在的
        if (id2Template.size()<=0){
            log.error("Can Not Acquire Template From TemplateClient:{}",request.getTemplateSDK().getId());
            throw new CouponException("Can Not Acquire Template From TemplateClient");
        }
        //判断用户是否可以领取这个优惠券
        List<Coupon> userUsableCoupons = findCouponsByStatus(request.getUserId(),CouponStatus.USABLE.getCode());
        Map<Integer, List<Coupon>> templateId2Coupons = userUsableCoupons.stream().collect(Collectors.groupingBy(Coupon::getTemplateId));

        if (templateId2Coupons.containsKey(request.getTemplateSDK().getId())&&
                templateId2Coupons.get(request.getTemplateSDK().getId()).size() >=
                        request.getTemplateSDK().getTemplateRule().getLimitation()){
            log.error("Exceed Template Assign Limitation:{}",request.getTemplateSDK().getId());
            throw new CouponException("Exceed Template Assign Limitation");
        }
        //尝试去获取优惠券码
        String couponCode= redisService.tryToAcquireCouponCodeFromCache(request.getTemplateSDK().getId());
        if (StringUtils.isEmpty(couponCode)){
            log.error("Can Not Acquire Coupon Code:{}",request.getTemplateSDK().getId());
            throw new CouponException("Can Not Acquire Coupon Code");
        }

        Coupon newCoupon = new Coupon(
                request.getTemplateSDK().getId(),
                request.getUserId(),
                couponCode,
                CouponStatus.USABLE
                );
        newCoupon = couponDao.save(newCoupon);

        //填充Coupon对象的CouponTemplateSDK，一定要在放入缓存之前填充
        newCoupon.setCouponTemplateSDK(request.getTemplateSDK());

        //放入缓存中
        redisService.addCouponToCache(request.getUserId(),Collections.singletonList(newCoupon),CouponStatus.USABLE.getCode());

        return newCoupon;
    }

    /**
     * 结算（核销）优惠券
     *
     * @param info {@link SettlementInfo}
     * @return {@link SettlementInfo}
     * @throws CouponException
     */
    @Override
    public SettlementInfo settlement(SettlementInfo info) throws CouponException {
        return settlementClient.computeRule(info).getData();
    }

    public static void main(String[] args) {
    }
}
